package com.atguigu.gmall.realtime.dim.app;/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/14 10:38
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.dim.function.DimSinkFunction;
import com.atguigu.gmall.realtime.dim.function.TableProcessFunction;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.bean.TableProcessDim;
import com.group2.edu.realtime.common.constant.Constant;
import com.group2.edu.realtime.common.util.FlinkSourceUtil;
import com.group2.edu.realtime.common.util.HBaseUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/14 10:38
 */
public class DimApp extends BaseApp {
    public static void main(String[] args) {
        new DimApp().start(10002,4,"dim_app",Constant.TOPIC_DB);

    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDs) {
        //TODO 1.对流中数据进行类型转换并进行简单的ETL(数据清洗), jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaStrDs.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                            String db = jsonObject.getString("database");
                            String type = jsonObject.getString("type");
                            String data = jsonObject.getString("data");
                            if ("gmall".equals(db)
                                    && ("insert".equals(type)
                                    || "update".equals(type)
                                    || "delete".equals(type)
                                    || "bootstrap-insert".equals(type))
                                    && data != null
                                    && data.length() > 2
                            ) {
                                out.collect(jsonObject);
                            }
                        } catch (Exception e) {
                            System.out.println("不是一个标准的json");
                        }
                    }
                }
        );
        // jsonObjDs.print();


        //TODO 2.使用FlinkCDC读取配置表中的配置信息
        //创建MySqlSource对象
        MySqlSource<String> mySqlSource = FlinkSourceUtil.getCDCMysqlSource("table_process_dim");
        DataStreamSource<String> mysqlStrDs = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL_Source");
        //mysqlStrDs.print();
        //"op":"r" {"before":null,"after":{"source_table":"financial_sku_cost","sink_table":"dim_financial_sku_cost","sink_family":"info","sink_columns":"id,sku_id,sku_name,busi_date,is_lastest,sku_cost,create_time","sink_row_key":"id"},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"gmall0620_config","sequence":null,"table":"table_process_dim","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1732870578960,"transaction":null}
        //"op":"c" {"before":null,"after":{"source_table":"t_a","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name","sink_row_key":"id"},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1732870643000,"snapshot":"false","db":"gmall0620_config","sequence":null,"table":"table_process_dim","server_id":1,"gtid":null,"file":"mysql-bin.000002","pos":11499579,"row":0,"thread":14,"query":null},"op":"c","ts_ms":1732870642929,"transaction":null}
        //"op":"u" {"before":{"source_table":"t_a","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name","sink_row_key":"id"},"after":{"source_table":"t_a","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name,age","sink_row_key":"id"},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1732870709000,"snapshot":"false","db":"gmall0620_config","sequence":null,"table":"table_process_dim","server_id":1,"gtid":null,"file":"mysql-bin.000002","pos":11499941,"row":0,"thread":14,"query":null},"op":"u","ts_ms":1732870709291,"transaction":null}
        //"op":"d" {"before":{"source_table":"t_a","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name,age","sink_row_key":"id"},"after":null,"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1732870748000,"snapshot":"false","db":"gmall0620_config","sequence":null,"table":"table_process_dim","server_id":1,"gtid":null,"file":"mysql-bin.000002","pos":11500331,"row":0,"thread":14,"query":null},"op":"d","ts_ms":1732870748451,"transaction":null}
        SingleOutputStreamOperator<TableProcessDim> tpDs = mysqlStrDs.map(
                new MapFunction<String, TableProcessDim>() {
                    @Override
                    public TableProcessDim map(String s) throws Exception {
                        //将jsonStr转换为jsonObj
                        JSONObject jsonObject = JSON.parseObject(s);
                        String op = jsonObject.getString("op");
                        TableProcessDim tableProcessDim = null;
                        if ("d".equals(op)) {
                            //对配置表进行了删除操作，从before属性中获取删除前的信息
                            //getObject  用于将 JSON 对象直接转换为指定的 Java 对象类型
                            tableProcessDim = jsonObject.getObject("before", TableProcessDim.class);
                        } else {
                            tableProcessDim = jsonObject.getObject("after", TableProcessDim.class);
                        }
                        //补充操作类型
                        tableProcessDim.setOp(op);
                        return tableProcessDim;
                    }
                }
        ).setParallelism(1);
        //tpDs.print();


        //TODO 3.根据配置表中的配置信息在HBase中执行建表或删表操作
        tpDs = tpDs.map(
                new RichMapFunction<TableProcessDim, TableProcessDim>() {
                    private Connection hbaseConn;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeConnection(hbaseConn);

                    }

                    @Override
                    public TableProcessDim map(TableProcessDim tp) throws Exception {
                        String op = tp.getOp();
                        String sinkTable = tp.getSinkTable();
                        String[] families = tp.getSinkFamily().split(",");
                        if ("r".equals(op) || "c".equals(op)){
                            //从配置表中读取一条数据或者向配置表中添加一条数据， 执行建表操作
                            HBaseUtil.createTable(hbaseConn,Constant.HBASE_NAMESPACE,sinkTable,families);

                        }else if ("d".equals(op)){
                            //从配置表中删除了一条配置， 执行删表操作
                            HBaseUtil.deleteTable(hbaseConn,Constant.HBASE_NAMESPACE,sinkTable);
                        }else {
                            //对表中的某条数据进行了更新操作，先删表再建表
                            HBaseUtil.deleteTable(hbaseConn,Constant.HBASE_NAMESPACE,sinkTable);
                            HBaseUtil.createTable(hbaseConn, Constant.HBASE_NAMESPACE,sinkTable,families);
                        }
                        return tp;
                    }
                }
        );
        //tpDs.print();

        //TODO 4.广播配置流 broadcast
        MapStateDescriptor<String, TableProcessDim> mapStateDescriptor =
                new MapStateDescriptor<String, TableProcessDim>("mapStateDescriptor", String.class, TableProcessDim.class);
        BroadcastStream<TableProcessDim> broadcastDs = tpDs.broadcast(mapStateDescriptor);
        //TODO 5.将主流和广播流进行关联 connect
        BroadcastConnectedStream<JSONObject, TableProcessDim> connectDs = jsonObjDs.connect(broadcastDs);

        //TODO 6.对关联后的数据进行处理
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> dimDs = connectDs.process(
                new TableProcessFunction(mapStateDescriptor)
        );
        dimDs.print();
        //TODO 7.将维度数据同步到Hbase
        dimDs.addSink(new DimSinkFunction());

    }
}
